import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Arrays;
import java.util.Properties;

public class kfkConsumer2 {
    //消费者
    public static void main(String[] args) {
        // 配置项加载
        Properties properties = new Properties();
        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "123.56.187.176:9092");//连虚拟机
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");//key反序列化
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");//value反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"niit");//注册消费者
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");

        //构建Kafka的消费者
        KafkaConsumer<String,String> kafkaConsumer = new KafkaConsumer<String,String>(properties);
        //订阅topic 名字
        kafkaConsumer.subscribe(Arrays.asList("t15"));
        //循环消费
        while (true){
            ConsumerRecords<String,String> conRecord = kafkaConsumer.poll(5);//每隔5秒
            for(ConsumerRecord<String,String> consumer : conRecord){
                System.out.println("value is " + consumer.value());
            }
//            for(ConsumerRecord<String,String> consumer : conRecord)
//            {
//                System.out.println(consumer.offset()+" "+consumer.value());
//                System.out.println("---------------------------------");
//                System.out.println("Consumer 信息"+consumer);

            //}
        }


    }
}
